package ltd.dujiabao.activemq.jms;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import javax.jms.*;

/**
 * @author : DuJiabao
 * @Project : active-mq
 * @Time : 2021/5/4 23:24
 * @Desc :
 */
public class TransactionConsumer {
  private static final String ACTIVEMQ_URL = "tcp://192.168.233.133:61616";  // 服务器端口号，地址
  private static final String DESTINATION_NAME = "queue01";  // 目的地名称
  private ActiveMQConnectionFactory activeMQConnectionFactory;
  private Connection connection = null;
  private Session session = null;
  private MessageConsumer consumer = null;

  @Before
  public void before() throws JMSException {
    activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
    connection = activeMQConnectionFactory.createConnection();
    connection.start();
  }

  /**
   * 消费者正常自动提交
   *
   * @throws Exception
   */
  @Test
  public void test() throws Exception {
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue queue = session.createQueue(DESTINATION_NAME);
    MessageConsumer consumer = session.createConsumer(queue);
    consumer.setMessageListener((message) -> {
      if (message instanceof TextMessage) {
        TextMessage textMessage = (TextMessage) message;
        try {
          System.out.println("获得消息" + textMessage.getText());
        } catch (JMSException e) {
          e.printStackTrace();
        }
      }
    });
    System.in.read();
  }

  /**
   * 消费者自己提交事务
   *
   * @throws Exception
   */
  @Test
  public void test2() throws Exception {
    session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
    Queue queue = session.createQueue(DESTINATION_NAME);
    consumer = session.createConsumer(queue);
    while (true) {
      Message message = consumer.receive();
      if (message instanceof TextMessage) {
        TextMessage textMessage = (TextMessage) message;
        String text = textMessage.getText();
        System.out.println("获得消息" + text);
        session.commit();
      } else {
        break;
      }
    }
    //consumer.setMessageListener((message) -> {
    //  if (message instanceof TextMessage) {
    //    TextMessage textMessage = (TextMessage) message;
    //    try {
    //      String text = textMessage.getText();
    //      System.out.println("获得消息" + text);
    //      if ("1".equals(text))
    //        session.commit();  // 只有当文本是1的时候，才提交
    //    } catch (JMSException e) {
    //      e.printStackTrace();
    //    }
    //  }
    //});
    //Thread.sleep(10000);
  }

  @Test
  public void test3() throws JMSException {
    session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE); // 改成客户端手动签收
    Queue queue = session.createQueue(DESTINATION_NAME);
    consumer = session.createConsumer(queue);
    while (true) {
      Message message = consumer.receive();
      if (message instanceof TextMessage) {
        TextMessage textMessage = (TextMessage) message;
        String text = textMessage.getText();
        System.out.println("获得消息" + text);
        textMessage.acknowledge();  // 确认签收
      } else {
        break;
      }
    }
  }

  @After
  public void after() throws JMSException {
    // 关闭资源
    consumer.close();
    session.close();
    connection.close();
    System.out.println("资源关闭完毕");
  }

}
